package com.spider.dao.hbasefilter;

import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.util.ArrayList;
import java.util.List;


public class HBaseDAl extends HBaseFilterImp {
	public HBaseDAl() {
		super();
	}

	/**
	 * 可以从HBase数据库读出表中某个列族某一行的所有列
	 *
	 * @param tableName
	 *            表名
	 * @param rowkey
	 *            行键
	 * @param family
	 *            列族
	 * @return
	 */
	public List<Result> readSingleFormSingleFamily(String tableName, String rowkey, String family) {
		List<Result> list = null;
		ResultScanner scanner = null;
		try {
			Table table = connection.getTable(TableName.valueOf(tableName));
			Filter filter = new RowFilter(CompareFilter.CompareOp.EQUAL, new BinaryComparator(Bytes.toBytes(rowkey)));
			Scan scan = new Scan();
			scan.setFilter(filter);
			scan.addFamily(Bytes.toBytes(family));
			scanner = table.getScanner(scan);
			list = new ArrayList<Result>();
			for (Result rs : scanner) {
				list.add(rs);
			}
		} catch (Exception e) {
			scanner.close();
		}
		return list;
	}

	/**
	 * 专门用来读取给定学生的推荐试题信息
	 *
	 * @param rowkey
	 *            行键
	 * @return
	 */
	public List<Result> readSingleStudentRecommendTest(String rowkey) {
		List<Result> list = readSingleFormSingleFamily("Student", rowkey, "recommend");
		List<Result> listTest = new ArrayList<Result>();
		;
		ResultScanner scanner = null;
		try {
			for (Result rs : list) {
				Cell[] rawCells = rs.rawCells();
				for (int i = 0; i < rawCells.length; i++) {
					Cell cell = rawCells[i];
					Table table = connection.getTable(TableName.valueOf("Test1"));
					Filter filterTest = new RowFilter(CompareFilter.CompareOp.EQUAL,
							new BinaryComparator(CellUtil.cloneQualifier(cell)));
					Scan scan = new Scan();
					scan.setFilter(filterTest);
					scan.addFamily(Bytes.toBytes("base"));
					scanner = table.getScanner(scan);
					for (Result test : scanner) {
						listTest.add(test);
					}
				}
			}
		} catch (Exception e) {
			scanner.close();
		}
		return listTest;
	}

	/**
	 * 专门用来读取试题中的题干
	 *
	 * @return
	 */
	public List<Result> readTestsContent() {
		String tableName = "Test1";
		String qualifier = "content";
		List<Result> list = null;
		ResultScanner scanner = null;
		try {
			Table table = connection.getTable(TableName.valueOf(tableName));
			Filter filter = new QualifierFilter(CompareFilter.CompareOp.EQUAL,
					new BinaryComparator(Bytes.toBytes(qualifier)));
			Scan scan = new Scan();
			scan.setFilter(filter);
			scanner = table.getScanner(scan);
			list = new ArrayList<Result>();
			for (Result rs : scanner) {
				list.add(rs);
			}
		} catch (Exception e) {
			scanner.close();
		}
		return list;
	}

	/**
	 * 专门用来读取学生表一个学生对题本的试题rowkey
	 * @param rowkey
	 * @return
	 */
	public List<Result> readStudentCorrect(String rowkey) {
		String tableName = "Student";
		String family = "correct";
		List<Result> list = null;
		ResultScanner scanner = null;
		try {
			Table table = connection.getTable(TableName.valueOf(tableName));
			Scan scan = new Scan();
			scan.addFamily(Bytes.toBytes(family));
			Filter filter = new RowFilter(CompareFilter.CompareOp.EQUAL, new BinaryComparator(Bytes.toBytes(rowkey)));

			scan.setFilter(filter);
			scanner = table.getScanner(scan);
			list = new ArrayList<Result>();
			for (Result rs : scanner) {
				list.add(rs);
			}
		} catch (Exception e) {
			scanner.close();
		}
		return list;
	}

	/**
	 * 专门用来读取试题中的answerUrl
	 *
	 * @return
	 */
	public List<Result> readTestsAnswerUrl() {
		String tableName = "Test1";
		String qualifier = "answerUrl";
		List<Result> list = null;
		ResultScanner scanner = null;
		try {
			Table table = connection.getTable(TableName.valueOf(tableName));
			Filter filter = new QualifierFilter(CompareFilter.CompareOp.EQUAL,
					new BinaryComparator(Bytes.toBytes(qualifier)));
			Scan scan = new Scan();
			scan.setFilter(filter);
			scanner = table.getScanner(scan);
			list = new ArrayList<Result>();
			for (Result rs : scanner) {
				list.add(rs);
			}
		} catch (Exception e) {
			scanner.close();
		}
		return list;
	}

	/**
	 * 专门用来读取学生表一个学生对题本的试题rowkey
	 * @param rowkey
	 * @return
	 */
	public List<Result> readStudentWrong(String rowkey){
		String tableName = "Student";
		String family = "wrong";
		List<Result> list = null;
		ResultScanner scanner = null;
		try {
			Table table = connection.getTable(TableName.valueOf(tableName));
			Scan scan = new Scan();
			scan.addFamily(Bytes.toBytes(family));
			Filter filter = new RowFilter(CompareFilter.CompareOp.EQUAL, new BinaryComparator(Bytes.toBytes(rowkey)));

			scan.setFilter(filter);
			scanner = table.getScanner(scan);
			list = new ArrayList<Result>();
			for (Result rs : scanner) {
				list.add(rs);
			}
		} catch (Exception e) {
			scanner.close();
		}
		return list;
	}

	/**
	 * 读取学生收藏列族所有数据
	 * @return
	 */
	public List<Result> readCollection(){
		HBaseFilter hbf = new HBaseFilterImp();
		String tableName="Student";
		String family="collection";
		return hbf.FamilyFilter(tableName, family);
	}
	/**
	 * 读取所有学生登录账号
	 * @param value 账号
	 * @return
	 */
	public List<Result> readAccout(String value){
		List<Result> list = null;
		ResultScanner scanner = null;
		try {
			Table table = connection.getTable(TableName.valueOf("Student"));
			Scan scan = new Scan();
			scan.addColumn(Bytes.toBytes("base"), Bytes.toBytes("account"));
			Filter filter = new ValueFilter(CompareFilter.CompareOp.EQUAL, new BinaryComparator(Bytes.toBytes(value)));
			scan.setFilter(filter);
			scanner = table.getScanner(scan);
			list = new ArrayList<Result>();
			for (Result rs : scanner) {
				list.add(rs);
			}
		} catch (Exception e) {
			scanner.close();
		}
		return list;
	}
}
